package com.atguigu.edu.realtime220815.util;

import com.atguigu.edu.realtime220815.app.func.TransientSink;
import com.atguigu.edu.realtime220815.common.EduConfig;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @Classname ClickHouseUtils
 * @Description TODO
 * @Date 2023/2/16 14:00
 * @Created by lzx
 */
public class ClickHouseUtils {
    /**
     * 所有dws层的类处理完数据,将数据往ClickHouse中写的时候都会调用此方法,来获取需要的SinkFunction,而SinkFunction是一个带泛型的
     * 类,说白了我们要将不同流中的不同类型数据写入ClickHouse所以我们不知道处理的具体是啥东西,只能使用泛型
     *
     * @param sql 程序向ClickHouse写数据的具体sql
     * @param <T> 此泛型为流中数据的类型,也就是需要发送到ClickHouse中的数据类型
     */
    public static <T> SinkFunction<T> getClickHouseSinkFunction(String sql) {
        //Flink给我们提供了获取SinkFunction的类,叫JdbcSink,里面的sink方法可以返回我们需要的SinkFunction
        /*
        在这里我们使用的是sink的四个参数的重载方法(String sql , JdbcStatementBuilder , JdbcExecutionOptions,JdbcConnectionOptions)
        1.sql:就是具体在ClickHouse中执行的sql语句,由外部传入
        2.JdbcStatementBuilder:用来填补sql中的占位符
        3.JdbcExecutionOptions:执行中的一些参数配置
        4.JdbcConnectionOptions:连接ClickHouse的具体配置,例如驱动和Url
         */
        return JdbcSink.sink(
                sql,
                new JdbcStatementBuilder<T>() {
                    @Override
                    public void accept(PreparedStatement preparedStatement, T t) throws SQLException {
                        try {
                        /*
                        此方法是填补sql中占位符的方法,泛型T就是流中的数据类型,因为是通用方法,所以并不知道我们处理的具体是什么类
                        所以这里只能使用反射来判断
                        想获取类的Class有三种方法,这里只能用 类的对象.getClass()这种方法,然后通过Class类中的 .getDeclaredFields()
                        来获取本类中所有的数据构成的数组
                         */
                            Field[] fields = t.getClass().getDeclaredFields();
                            //定义一个变量来记录跳过属性的个数
                            int skipField = 0;
                            for (int i = 0; i < fields.length; i++) {
                                //遍历数组,获取每一个属性
                                Field field = fields[i];
                                //因为属性有可能为私有,通过setAccessible()为true就可以越过权限获取到值
                                field.setAccessible(true);
                                /*
                                有时,有的JavaBean里面的某些属性不存在与对应的ClickHouse表中,也就是说某些属性不需要,则给这些属性
                                打一个标记,也就是加一个注解,这里我们自定义注解TransientSink,通过属性的getAnnotation(TransientSink.class)
                                方法获取到TransientSink的对象,如果该对象不为空,则说明这个属性中有我们加上的注解,此属性不需要往
                                数据库里面写,则直接continue跳到下一轮的属性获取上
                                 */
                                TransientSink annotation = field.getAnnotation(TransientSink.class);
                                if(annotation!=null){
                                    //如果跳过了,则变量skipField ++
                                    skipField++;
                                    continue;
                                }
                                //通过属性的.get()方法,将对象t传入,就可以获取到t对象里面的这个属性的属性值
                                Object fieldValue = field.get(t);
                                /*
                                通过preparedStatement来将获取到的值填充到sql中的占位符里面,这里要注意,fields数组的遍历是
                                从0开始的,但是sql的占位符填充是从1开始的,所以第i位的属性要填充的占位符为第i+1位
                                但是有时会发生跳过属性(该属性不需要往CK中写)的场景,比如说第一第二个属性需要往CK中写,则
                                i    0  1
                                i+1  1  2
                                本来i为2的时候需要填充第三个占位符,但是现在i为2的属性不需要写入CK,也就不需要填写占位符,此时
                                i为3的属性填写i+1为3的占位符
                                解决方法是使用变量来记录属性跳过的个数,在填充站位的时候只需要减去跳过的个数及可以填充正确的站位符
                                skipField初始值为0
                                 */
                                preparedStatement.setObject(i+1-skipField,fieldValue);
                            }
                        } catch (IllegalAccessException e) {
                            throw new RuntimeException(e);
                        }
                    }
                },
                new JdbcExecutionOptions.Builder()
                        .withBatchSize(5)//攒够一批再往ClickHouse里面发,这里是同一并行度攒够五条数据才往ClickHouse发
                        .withBatchIntervalMs(3000)//如果隔了3000ms还没有攒够一批数据(这里是5条),就把现有的数据发往ClickHouse
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl(EduConfig.CLICKHOUSE_URL)//URl
                        .withDriverName(EduConfig.CLICKHOUSE_DRIVER)//驱动
                        .build()
        );
    }
}
